package com.hz.cheetah.client;

import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import com.hz.cheetah.common.bean.RpcRequest;
import com.hz.cheetah.common.bean.RpcResponse;
import com.hz.cheetah.common.sys.JvmRuntime;
import com.hz.cheetah.common.util.StringUtil;
import com.hz.cheetah.hub.ServiceDiscovery;
import com.hz.cheetah.hub.protocol.CheetahURL;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.security.auth.login.ConfigurationSpi;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;

/**
 * jdk 动态代理方式，执行速度根据观察不低于javassist 创建的代理类（真是怪哉。。。）
 *
 * @author chuanyang
 * @since 1.0.0
 */
public class CheetahByJDKProxy implements ServiceBeanFactory{

    private static final Logger LOGGER = LoggerFactory.getLogger(CheetahByJDKProxy.class);

    private static final String SUFFIX = "Async";

    final static ListeningExecutorService service = MoreExecutors.listeningDecorator(
            Executors.newFixedThreadPool(JvmRuntime.getRuntime().getCPUsNumber() * 2));


    private ServiceDiscovery serviceDiscovery;

    /**
     * 当操作请求失败,重新执行的次数
     */
    private int doCount = 3;

    public void setDoCount(int count){
        this.doCount = count;
    }

    public ServiceDiscovery getServiceDiscovery() {
        return serviceDiscovery;
    }

    public void setServiceDiscovery(ServiceDiscovery serviceDiscovery) {
        this.serviceDiscovery = serviceDiscovery;
    }

    public CheetahByJDKProxy(){

    }

    public CheetahByJDKProxy(ServiceDiscovery serviceDiscovery) {
        this.serviceDiscovery = serviceDiscovery;
    }

    @SuppressWarnings("unchecked")
    public <T> T create(final Class<?> interfaceClass) {
        return create(interfaceClass, "");
    }

    @SuppressWarnings("unchecked")
    public <T> T create(final Class<?> interfaceClass, final String serviceVersion) {
        // 创建动态代理对象
        return (T) Proxy.newProxyInstance(
                interfaceClass.getClassLoader(),
                new Class<?>[]{interfaceClass},
                new InvocationHandler() {
                    @Override
                    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                        // 创建 RPC 请求对象并设置请求属性
                        RpcRequest request = new RpcRequest();
                        request.setRequestId(UUID.randomUUID().toString());
                        String interfaceName = method.getDeclaringClass().getName();
                        /**
                         * 在调试模式下,调试框架有可能访问远程对象的toString 方法。
                         */
                        if(Object.class.getName().equals(interfaceName)){
                            return "cheetah remote object toString method invoked";
                        }

                        request.setServiceVersion(serviceVersion);
                        request.setParameterTypes(method.getParameterTypes());
                        request.setParameters(args);
                        String methodName = method.getName();
                        String interfaceClassName = interfaceClass.getName();
                        if(interfaceClassName.endsWith(SUFFIX)){
                            interfaceClassName = interfaceClassName.substring(0,interfaceClassName.length()-SUFFIX.length());
                            if(methodName.endsWith(SUFFIX)){
                                request.setType(RpcRequest.TYPE_ASYNC);
                                methodName = methodName.substring(0,methodName.length() - SUFFIX.length());
                            }else{
                                request.setType(RpcRequest.TYPE_RPC);
                            }
                        }
                        request.setInterfaceName(interfaceClassName);
                        request.setMethodName(methodName);

                        String serviceName = interfaceClassName;
                        if (StringUtil.isNotEmpty(serviceVersion)) {
                            serviceName += "-" + serviceVersion;
                        }
                        if(request.getType() == RpcRequest.TYPE_ASYNC){
                            final String serName = serviceName;
                            return service.submit(new Callable<Object>() {
                                @Override
                                public Object call() throws Exception {
                                    return synchronizeExe(request, serName);
                                }
                            });
                        }else{
                            return synchronizeExe(request, serviceName);
                        }

                    }
                }
        );
    }

    public Object synchronizeExe(RpcRequest request, String serName) {
        RpcResponse response = execute(request, serName, null, 0, null);
        if (response == null) {
            throw new RuntimeException("response is null");
        }
        // 返回 RPC 响应结果
        if (response.hasException()) {
            throw new RuntimeException(response.getException());
        } else {
            return response.getResult();
        }
    }

    private RpcResponse execute(RpcRequest request, String serviceName, CheetahURL old, int num, Exception e) {
        if (num < doCount) {
            CheetahURL serviceAddress = getServiceAddress(serviceName, old);
            CheetahClient client = getClient(serviceAddress);
            try {
                RpcResponse response = client.send(request);
                if (response == null) {
                    num += 1;
                    LOGGER.warn("requestid {}:retry num:{}",request.getRequestId(), num);
                    return execute(request, serviceName, serviceAddress, num, null);
                }
                return response;
            } catch (Exception e1) {
                num += 1;
                LOGGER.warn("retry num:{}", num);
                return execute(request, serviceName, serviceAddress, num, e1);
            }
        } else {
            RpcResponse response = new RpcResponse();
            response.setRequestId(request.getRequestId());
            response.setException(e);
            return response;
        }
    }

    public CheetahClient getClient(CheetahURL serviceAddress) {
        String address = serviceAddress.getAdress();
        CheetahClient client = ClientManager.channelMap.get(address);//new CheetahClient(serviceAddress);
        if (client == null) {
            synchronized (address) {
                client = ClientManager.channelMap.get(address);
                if (client == null) {
                    client = new CheetahClient(serviceAddress);
                    ClientManager.channelMap.put(address, client);
                }
            }
        }
        address = null;
        return client;
    }

    protected CheetahURL getServiceAddress(String serviceName, CheetahURL old) {
        CheetahURL serviceAddress = null;
        if (serviceDiscovery != null) {
            if (old == null) {
                serviceAddress = serviceDiscovery.discover(serviceName);
            } else {
                serviceAddress = serviceDiscovery.reDiscover(serviceName, old);
            }

            LOGGER.debug("discover service: {} => {}", serviceName, serviceAddress.getAdress());
        }
        if (serviceAddress == null) {
            throw new RuntimeException("server address is empty");
        }
        return serviceAddress;
    }
}
